package driver

import (
	"context"
	"errors"
	"gitee.com/huhong/netbolt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"io"
	"time"
)

const (
	address     = "localhost:9995"
	defaultName = "world"
)

var (
	NOT_LOGON = errors.New("Not found token!")

	//connTokens = &sync.Map{}
)

type Connection struct {
	Options      []grpc.CallOption
	Connected    bool
	DriverOption *driverOptions
	client       netbolt.NetBoltServiceClient
	grpcConn     *grpc.ClientConn
	token        string
	id           string
}

type driverOptions struct {
	ConnectTimeout time.Duration
	CallTimeout    time.Duration
	DialOptions    []grpc.DialOption
	CallOptions    []grpc.CallOption
}

func NewDriverOptions() driverOptions {
	return driverOptions{ConnectTimeout: 3 * time.Second}
}

func Connect(server string, username string, pwd string, dbName string, co *driverOptions) (*Connection, error) {
	connId := netbolt.UUID()

	conn := &Connection{Options: co.CallOptions, DriverOption: co, Connected: true, id: connId}
	opts := append(co.DialOptions, grpc.WithInsecure())
	unaryInterceptor := grpc.WithUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		if method == "/netbolt.NetBoltService/Connect" {
			return invoker(ctx, method, req, reply, cc, opts...)
		}

		//_, ok := connTokens.Load(connId)

		if conn.token != "" {
			md := metadata.Pairs("token", conn.token)

			newCtx := metadata.NewOutgoingContext(ctx, md)
			return invoker(newCtx, method, req, reply, cc, opts...)
		} else {
			return NOT_LOGON
		}

	})

	streamInterceptor := grpc.WithStreamInterceptor(func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		if conn.token != "" {
			md := metadata.Pairs("token", conn.token)

			newCtx := metadata.NewOutgoingContext(ctx, md)
			return streamer(newCtx, desc, cc, method, opts...)
		} else {
			return nil, NOT_LOGON
		}
	})
	opts = append(opts, unaryInterceptor)
	opts = append(opts, streamInterceptor)
	grpcConn, err := grpc.Dial(server, opts...)

	if err != nil {
		return nil, err
	}

	grpcClient := netbolt.NewNetBoltServiceClient(grpcConn)
	//callOptions = []grpc.CallOption{grpc.UseCompressor(gzip.Name)}
	ctx, cancel := context.WithTimeout(context.Background(), co.ConnectTimeout)
	//md := metadata.Pairs("myid", "client")
	//ctx = metadata.NewOutgoingContext(ctx, md)

	//grpc.WithStreamInterceptor(func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
	//
	//})

	result, err := grpcClient.Connect(ctx, &netbolt.LoginRequest{Username: username, Password: pwd, Db: dbName}, co.CallOptions...)
	defer cancel()
	if err != nil {
		return nil, err
	}

	//token: result.Token, client: grpcClient, grpcConn: grpcConn,

	//connTokens.Store(connId, result.Token)
	conn.token = result.Token
	conn.client = grpcClient
	conn.grpcConn = grpcConn
	return conn, nil
}

func (c *Connection) Close() error {
	resp, err := c.client.Close(context.Background(), &netbolt.Ticket{Token: c.token}, c.Options...)
	if err != nil {
		return err
	}
	if resp.Status != 0 {
		return errors.New(resp.Msg)
	}
	c.grpcConn.Close()
	return nil
}

func (c *Connection) Put(key []byte, value []byte) error {
	resp, err := c.client.Put(context.Background(), &netbolt.PutRequest{Key: key, Value: value}, c.Options...)
	if err != nil {
		return err
	}
	if resp.Status != 0 {
		return errors.New(resp.Msg)
	}
	return nil
}

func (c *Connection) createContext() (context.Context, context.CancelFunc) {
	var ctx context.Context
	var cancel context.CancelFunc
	if c.DriverOption.CallTimeout.Nanoseconds() == 0 {
		ctx = context.Background()
	} else {
		ctx, cancel = context.WithTimeout(context.Background(), c.DriverOption.CallTimeout)

	}
	return ctx, cancel
}

func (c *Connection) AsyncPut(key []byte, value []byte, callback chan error) {
	//如果不限制超时则无限等待这里应当考虑到超时

	ctx, cancel := c.createContext()
	resp, err := c.client.Put(ctx, &netbolt.PutRequest{Key: key, Value: value}, c.Options...)
	if cancel != nil {
		defer cancel()
	}

	if err == nil {
		if resp == nil {
			err = ctx.Err()
		} else {
			if resp.Status != 0 {
				err = errors.New(resp.Msg)
			}
		}
	}

	callback <- err
}

func (c *Connection) BatchPut(puts []*netbolt.Entity) error {

	resp, err := c.client.BatchPut(context.Background(), &netbolt.PutsRequest{Puts: puts}, c.Options...)
	if err != nil {
		return err
	}
	if resp.Status != 0 {
		return errors.New(resp.Msg)
	}
	return nil
}

func (c *Connection) AsyncBatchPut(puts []*netbolt.Entity, callback chan error) {
	ctx, cancel := c.createContext()
	resp, err := c.client.BatchPut(context.Background(), &netbolt.PutsRequest{Puts: puts}, c.Options...)
	if cancel != nil {
		defer cancel()
	}

	if err == nil {
		if resp == nil {
			err = ctx.Err()
		} else {
			if resp.Status != 0 {
				err = errors.New(resp.Msg)
			}
		}
	}

	callback <- err
}

func (c *Connection) Stream(op netbolt.StreamOp, handle func(entry *netbolt.Entry) (*netbolt.StreamOp, error)) error {

	//op.Token = c.token
	s, err := c.client.Stream(context.Background(), c.Options...)

	if err != nil {
		return err
	}

	err = s.Send(&op)

	if err != nil {

		return err
	}
	for {
		in, err := s.Recv()
		if err == io.EOF {
			return nil
		}

		if err != nil {
			s.CloseSend()
			return err
		}

		op, err := handle(in)

		if err != nil {
			s.CloseSend()
			return err
		}

		if op == nil {
			s.CloseSend()
			return nil
		}

		s.Send(op)
	}

}

func (c *Connection) Get(key []byte) ([]byte, error) {
	r, err := c.client.Get(context.Background(), &netbolt.GetRequest{Key: key}, c.Options...)
	if err != nil {
		return nil, err
	}
	return r.Value, nil
}

func (c *Connection) Delete(key []byte) error {
	resp, err := c.client.Delete(context.Background(), &netbolt.DeleteRequest{Key: key}, c.Options...)

	if err != nil {
		return err
	}
	if resp.Status != 0 {
		return errors.New(resp.Msg)
	}
	return nil
}

func (c *Connection) BatchDelete(keys [][]byte) error {
	resp, err := c.client.BatchDelete(context.Background(), &netbolt.BatchDeleteRequest{Keys: keys}, c.Options...)
	if err != nil {
		return err
	}
	if resp.Status != 0 {
		return errors.New(resp.Msg)
	}
	return nil
}

func (c *Connection) AsyncDelete(key []byte, callback chan error) {
	ctx, cancel := c.createContext()
	resp, err := c.client.Delete(ctx, &netbolt.DeleteRequest{Key: key}, c.Options...)
	if cancel != nil {
		defer cancel()
	}

	if err == nil {
		if resp == nil {
			err = ctx.Err()
		} else {
			if resp.Status != 0 {
				err = errors.New(resp.Msg)
			}
		}
	}
	callback <- err
}

func (c *Connection) AsyncBatchDelete(keys [][]byte, callback chan error) {
	ctx, cancel := c.createContext()
	resp, err := c.client.BatchDelete(ctx, &netbolt.BatchDeleteRequest{Keys: keys}, c.Options...)
	if cancel != nil {
		defer cancel()
	}
	if err == nil {
		if resp == nil {
			err = ctx.Err()
		} else {
			if resp.Status != 0 {
				err = errors.New(resp.Msg)
			}
		}
	}
	callback <- err
}

func CRC_CCITT(byteArray []byte) uint16 {
	var crc uint16
	for i := 0; i < len(byteArray); i++ {
		b := uint16(byteArray[i])
		q := (crc ^ b) & 0x0f
		crc = (crc >> 4) ^ (q * 0x1081)
		q = (crc ^ (b >> 4)) & 0xf
		crc = (crc >> 4) ^ (q * 0x1081)
	}
	return (crc >> 8) ^ (crc << 8)
}

func CalculateSlot(key []byte, maxSlot int) int {
	return int(CRC_CCITT(key)) % maxSlot
}

//func init() {
//
//	conn, err := grpc.Dial(address, grpc.WithBlock(), grpc.WithInsecure())
//	if err != nil {
//		log.Fatalf("did not connect: %v", err)
//	}
//
//	client = netbolt.NewNetBoltServiceClient(conn)
//	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
//
//	r, err := client.Connect(ctx, &netbolt.LoginRequest{Username: "huhong", Password: "123", Db: "sys"}, callOptions...)
//	defer cancel()
//
//	if err != nil {
//		log.Fatalf("could not connect:%v\n", err)
//	}
//
//	tickt = r.Token
//
//}
//
//func Put(key []byte, value []byte) {
//	ctx, _ := context.WithTimeout(context.Background(), 20*time.Second)
//
//	_, err := client.Put(ctx, &netbolt.PutRequest{Key: key, Value: value, Token: tickt}, callOptions...)
//
//	//defer cancel()
//
//	if err != nil {
//		log.Fatalf("could not put:%v\n", err)
//	}
//
//}
//
//func BatchPut(entity []*netbolt.Entity) {
//	ctx, _ := context.WithTimeout(context.Background(), 20*time.Second)
//
//	param := &netbolt.PutsRequest{Puts: entity, Token: tickt}
//	_, err := client.BatchPut(ctx, param, callOptions...)
//
//	if err != nil {
//		log.Fatalf("could not batchput:%v\n", err)
//	}
//}
//func Get(key []byte) ([]byte, error) {
//	ctx, _ := context.WithTimeout(context.Background(), time.Second)
//	r, err := client.Get(ctx, &netbolt.GetRequest{Key: key, Token: tickt}, callOptions...)
//
//	//defer cancel()
//
//	if err != nil {
//		log.Fatalf("could not put:%v\n", err)
//		return nil, err
//	}
//
//	return r.Value, nil
//
//}
//
//type NetBoltCursor netbolt.Cursor
//type Key struct {
//	Key []byte
//}
//
////func CreateCursor() (*NetBoltCursor, error) {
////	ticket := &netbolt.Ticket{Token: tickt}
////	ctx, _ := context.WithTimeout(context.Background(), time.Second)
////	c, err := client.CreateCursor(ctx, ticket, callOptions...)
////
////	if err != nil {
////		return nil, err
////	}
////	var nc = NetBoltCursor(*c)
////
////	return &nc, err
////}
//
////func (c *NetBoltCursor) Next() (*Key, error) {
////
////	cc := netbolt.Cursor(*c) //&netbolt.Cursor{Id: c.Id, Token: c.Token}
////	ctx, _ := context.WithTimeout(context.Background(), time.Second)
////	e, err := client.Next(ctx, &cc, callOptions...)
////
////	if err != nil {
////		return nil, err
////	} else {
////		if e != nil && e.Key != nil {
////			return &Key{e.Key}, nil
////		} else {
////			return nil, nil
////		}
////
////	}
////}
//
//func Size() int64 {
//	r, _ := client.Count(context.Background(), &netbolt.Ticket{Token: tickt}, callOptions...)
//
//	return r.Size
//}
//func Delete(key []byte) error {
//	_, err := client.Delete(context.Background(), &netbolt.DeleteRequest{Key: key, Token: tickt}, callOptions...)
//	return err
//}
//func BatchDelete(keys [][]byte) error {
//	_, err := client.BatchDelete(context.Background(), &netbolt.BatchDeleteRequest{Keys: keys, Token: tickt}, callOptions...)
//	return err
//}
//func Stream(op *netbolt.StreamOp, handler func(entry *netbolt.Entry) (*netbolt.StreamOp, error)) error {
//
//	s, err := client.Stream(context.Background(), callOptions...)
//
//	if err != nil {
//		return err
//	}
//
//	err = s.Send(op)
//
//	if err != nil {
//
//		return err
//	}
//	for {
//		in, err := s.Recv()
//		if err == io.EOF {
//			return nil
//		}
//
//		if err != nil {
//			s.CloseSend()
//			return err
//		}
//
//		newop, err := handler(in)
//
//		if err != nil {
//			s.CloseSend()
//			return err
//		}
//
//		if newop == nil {
//			s.CloseSend()
//			return nil
//		}
//
//		s.Send(newop)
//	}
//
//}
//
//func (k *Key) GetValue() ([]byte, error) {
//	return Get(k.Key)
//}
//
//func (c *NetBoltCursor) Seek() {
//
//}
